package com.xm.danmaku.service;


import com.xm.danmaku.dto.DanmakuMessage;

import java.util.concurrent.*;
import java.util.function.Predicate;

/**
 * 弹幕消息发布中心（基于Flow API）
 */
public class DanmakuPublisher {
    // 使用单例模式
    private static final DanmakuPublisher INSTANCE = new DanmakuPublisher();

    // 使用SubmissionPublisher作为基础实现
    private final SubmissionPublisher<DanmakuMessage> publisher;
    // 房间过滤器缓存
    private final ConcurrentHashMap<String, Predicate<DanmakuMessage>> roomFilters;
    // 线程池
    private final ExecutorService executor;

    private DanmakuPublisher() {
        this.executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
        this.publisher = new SubmissionPublisher<>(executor, Flow.defaultBufferSize());
        this.roomFilters = new ConcurrentHashMap<>();
    }

    public static DanmakuPublisher getInstance() {
        return INSTANCE;
    }

    /**
     * 发布新弹幕
     */
    public void publish(DanmakuMessage danmaku) {
        publisher.submit(danmaku);
    }

    /**
     * 订阅指定房间的弹幕
     */
    public Flow.Subscription subscribe(String roomId, Flow.Subscriber<DanmakuMessage> subscriber) {
        // 创建房间过滤器
        Predicate<DanmakuMessage> filter = msg -> roomId.equals(msg.getRoomId());
        roomFilters.putIfAbsent(roomId, filter);

        // 创建过滤处理器
        FilterProcessor processor = new FilterProcessor(filter, executor);
        publisher.subscribe(processor);
        processor.subscribe(subscriber);

        return new DanmakuSubscription(processor);
    }

    /**
     * 自定义Subscription用于取消订阅
     */
    private static class DanmakuSubscription implements Flow.Subscription {
        private final FilterProcessor processor;

        DanmakuSubscription(FilterProcessor processor) {
            this.processor = processor;
        }

        @Override
        public void request(long n) {
            // 不实现背压控制
        }

        @Override
        public void cancel() {
            processor.cancel();
        }
    }

    /**
     * 自定义Processor实现房间过滤
     */
    private static class FilterProcessor extends SubmissionPublisher<DanmakuMessage>
            implements Flow.Processor<DanmakuMessage, DanmakuMessage> {

        private final Predicate<DanmakuMessage> filter;
        private Flow.Subscription subscription;

        FilterProcessor(Predicate<DanmakuMessage> filter, ExecutorService executor) {
            super(executor, Flow.defaultBufferSize());
            this.filter = filter;
        }

        @Override
        public void onSubscribe(Flow.Subscription subscription) {
            this.subscription = subscription;
            subscription.request(Long.MAX_VALUE); // 不限制请求数量
        }

        @Override
        public void onNext(DanmakuMessage item) {
            if (filter.test(item)) {
                submit(item); // 转发给订阅者
            }
        }

        @Override
        public void onError(Throwable throwable) {
            closeExceptionally(throwable);
        }

        @Override
        public void onComplete() {
            close();
        }

        public void cancel() {
            if (subscription != null) {
                subscription.cancel();
            }
            close();
        }
    }
}
